/*******************************************************************************
 * Copyright (c) 2012, 2017 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *   http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial contribution
 *******************************************************************************/

#include "MQTTClient.h"
#include "cJSON.h"
#include <arpa/inet.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#define ADDRESS "tcp://81.70.2.209:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC "WindowsToLinux"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L

#define PORT 8888
#define IP "192.168.250.100"
#define PRINT_ERR(s)                         \
    do {                                     \
        fprintf(stderr, "__%d__", __LINE__); \
        perror(s);                           \
        return -1;                           \
    } while (0)

volatile MQTTClient_deliveryToken deliveredtoken;
int mqtt_init();
int tcp_init();
int get_newfd();
void recv_handler(void* args);

MQTTClient client;
int sfd, newfd;
void delivered(void* context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void* context, char* topicName, int topicLen, MQTTClient_message* message)
{
    int i, res;
    char* payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = message->payload;
    for (i = 0; i < message->payloadlen; i++) {
        putchar(*payloadptr++);
    }
    putchar('\n');

    cJSON* json = cJSON_Parse(message->payload);
    // 将传入的JSON结构转化为字符串 并打印
    char* json_data = NULL;
    printf("data:%s\n", json_data = cJSON_Print(json));
    res = send(newfd, message->payload, strlen(message->payload), 0);
    printf("newfd= %d\tres = %d\n", newfd, res);
    if (res < 0) {
        printf("设备断线!等待设备重连。。。\n");
        return -1;
    }
    free(json_data);
    // 将JSON结构所占用的数据空间释放
    cJSON_Delete(json);

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void* context, char* cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int main(int argc, char* argv[])
{
    char ch;
    pthread_t tid;
    // mqtt初始化
    if (0 != mqtt_init()) {
        puts("init_mqtt err");
        return -1;
    }
    if (0 != tcp_init())
        return -1;
    pthread_create(&tid, NULL, (void*)recv_handler, NULL);

    do {
        ch = getchar();
    } while (ch != 'Q' && ch != 'q');

    MQTTClient_unsubscribe(client, TOPIC);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    close(sfd);
    close(newfd);
    return 0;
}

void recv_handler(void* args)
{
    char buf[64] = "";
    int res = -1;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    while (1) {
        // 接收信息
        bzero(buf, sizeof(buf));
        res = recv(newfd, buf, sizeof(buf), 0);
        if (res < 0) {
            perror("recv");
            continue;
        }
        if (0 == res) {
            printf("newfd=%d 客户端退出！\n", newfd);
            get_newfd();
            continue;
        }
        printf("%s\n", buf);
        pubmsg.payload = buf;
        pubmsg.payloadlen = (int)strlen(buf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client, "LinuxToWindows", &pubmsg, &token);
    }
}
int tcp_init()
{
    // 创建流式套接字
    sfd = socket(AF_INET, SOCK_STREAM, 0);
    if (sfd < 0)
        PRINT_ERR("socket");

    // 允许端口快速复用
    int reuse = 1;
    if (setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0)
        PRINT_ERR("setsockopt");

    // 填充IP信息并绑定
    struct sockaddr_in sin;
    sin.sin_addr.s_addr = inet_addr(IP);
    sin.sin_family = AF_INET;
    sin.sin_port = htons(PORT);
    if (bind(sfd, (struct sockaddr*)&sin, sizeof(sin)) < 0)
        PRINT_ERR("bind");

    // 设置监听
    if (listen(sfd, 128) < 0)
        PRINT_ERR("listen");
    get_newfd();

    return 0;
}

int mqtt_init()
{
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int ch;

    MQTTClient_create(&client, ADDRESS, CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
        TOPIC, CLIENTID, QOS);
    MQTTClient_subscribe(client, TOPIC, QOS);
}

int get_newfd()
{
    struct sockaddr_in cin;
    socklen_t cinlen = sizeof(cin);
    newfd = -1;
    // 获取新的套接字
    newfd = accept(sfd, (struct sockaddr*)&cin, &cinlen);
    if (newfd < 0)
        PRINT_ERR("accept");
    printf("[%s:%d]newfd=%d 连接成功！\n", inet_ntoa(cin.sin_addr), ntohs(cin.sin_port), newfd);
    return 0;
}